//
// Created by dongbo on 11/12/18.
//
#include "message.h"
#include "pnode.h"
#include "../z_hglog.h"
#include "evpaxos.h"

#include <cassert>
#include <stdlib.h>
#include <stdio.h>
#include <atomic>
#include <mutex>
#include <sstream>
#include <vector>
#include <thread>
#include <netinet/tcp.h>
#include <signal.h>
#include <unistd.h>
#include <cstring>
#include <sys/prctl.h>


void pclient::client_submit_value(struct client *c, std::string str_msg) {
    char ss[4096] = {0};
    auto v = reinterpret_cast<client_value *>(ss);
    v->size = str_msg.length();
    v->client_id = c->id;
    gettimeofday(&v->t, nullptr);
    sprintf(v->value, "%s", str_msg.c_str());
    paxos_submit(c->bev, ss, sizeof(struct client_value) + str_msg.length());
}


void pclient::on_connect(struct bufferevent *bev, short events, void *arg) {
    auto pc = reinterpret_cast<struct pclient *>(arg);
    auto c = pc->c;
    if (events & BEV_EVENT_CONNECTED) {
        int flag = 1;
        setsockopt(bufferevent_getfd(pc->c->bev), IPPROTO_TCP, TCP_NODELAY, &flag, sizeof(int));

//        t_paxos_msg::lease tp;
//        tp.set_msg(c->id, time(0));
//        client_submit_value(c, tp.to_string());

        pc->ok = true;
        T_LOGI("client have Connected to proposer");
    } else if (events & (BEV_EVENT_TIMEOUT | BEV_EVENT_ERROR | BEV_EVENT_EOF)) {
        event_add(c->timer, &c->tv);
    } else {
        printf("%s\n", evutil_socket_error_to_string(EVUTIL_SOCKET_ERROR()));
    }
}


void pclient::on_pipe_read(struct bufferevent *bevt, void *arg) {
    auto pc = reinterpret_cast<struct pclient *>(arg);
    auto c = pc->c;

    unsigned long long msg1[1000] = {0};
    auto len = bufferevent_read(bevt, (void *) msg1, 1000 * sizeof(unsigned long long));
    for (int i = 0; i < len / sizeof(unsigned long long); ++i) {
        auto ms = reinterpret_cast<msg *>(msg1[i]);
        client_submit_value(c, ms->str);
        delete ms;
    }
}

void pclient::on_connection_err(int fd, short ev, void *arg) {
    auto pc = reinterpret_cast<pclient *>(arg);
    pc->connect_to_proposer(pc, pc->appid, pc->config);
}

void pclient::connect_to_proposer(struct pclient *pc, int proposer_id, struct evpaxos_config *conf) {
    if (conf == nullptr) {
        printf("Failed to read config\n");
        return;
    }
    struct sockaddr_in addr = evpaxos_proposer_address(reinterpret_cast<struct evpaxos_config *>(conf), proposer_id);

    if (pc->c->bev) bufferevent_free(pc->c->bev);
    pc->c->bev = bufferevent_socket_new(pc->c->base, -1, BEV_OPT_CLOSE_ON_FREE);
    if (pc->c->bev == NULL) exit(1);

    bufferevent_setcb(pc->c->bev, NULL, NULL, on_connect, (void *) pc);
    bufferevent_enable(pc->c->bev, EV_READ | EV_WRITE);
    bufferevent_socket_connect(pc->c->bev, (struct sockaddr *) &addr, sizeof(addr));

}

void pclient::make_client(int id) {
    c = reinterpret_cast<client *>(malloc(sizeof(client)));
    c->bev = nullptr;
    c->bev0 = nullptr;
    c->timer = nullptr;
    c->base = nullptr;
    c->id = id;
    c->base = event_base_new();
    c->tv.tv_sec = 5;
    c->tv.tv_usec = 0;
    c->timer = evtimer_new(c->base, on_connection_err, this);

    new std::thread([this, id]() {
        prctl(PR_SET_NAME, "pclient");
        {
            connect_to_proposer(this, appid, config);
        }

        {
            int pipe_fd[2];
            pipe(pipe_fd);

            c->bev0 = bufferevent_socket_new(c->base, pipe_fd[0], BEV_OPT_CLOSE_ON_FREE);
            bufferevent_setcb(c->bev0, on_pipe_read, NULL, NULL, this);
            bufferevent_enable(c->bev0, EV_READ);
            c->write_pipefd = pipe_fd[1];

            sigset_t sset;
            sigemptyset(&sset);
            sigaddset(&sset, SIGPIPE);
            pthread_sigmask(SIG_BLOCK, &sset, NULL);
        }

        event_base_dispatch(c->base);
    });
}

void pclient::client_free() {
    bufferevent_free(c->bev);
    event_free(c->stats_ev);
    event_free(c->sig);
    event_base_free(c->base);
    if (c->learner)
        evlearner_free(c->learner);
    free(c);
}

void pclient::start_client() {
    make_client(appid);
}

int pclient::SendMsg(std::string s, int id) {
    auto ms = new msg;
    ms->str = s;
    ms->id = id;
    auto a = (unsigned long long) ms;
    auto f = write(c->write_pipefd, (char *) &a, sizeof(unsigned long long));
    assert(f >= 8);
    return 0;
}

//==========================================================================================================

pnode::pnode(int _id,
             struct evpaxos_config *_config,
             zmqpipe *_zpip,
             ppax_cb *_ppxcb) :
        id(_id),
        config(_config),
        num_pxnode(_config->acceptors_count) {

    client_node = new pclient(id, config);
    proposer = new role_proposer(_ppxcb);
    acceptor = new role_acceptor(_ppxcb);
    learner = new role_learner(id, acceptor, proposer, _ppxcb, config);
}


int pnode::init_pnode() {
    acceptor->start_acceptor();
    proposer->start_proposer();
    learner->start_learner();

    //----------------------------------------------
    client_node->start_client();

    return 0;
}


int pnode::proposer_connect_new_acceptor(int id, struct sockaddr_in) {
    return 0;
}

int pnode::leaner_connect_new_acceptor(int id, struct sockaddr_in) {
    return 0;
}

void pnode::set_zpip(zmqpipe *_zpip, int zpip_num) {
    if (learner && _zpip) {
        learner->set_zpip(_zpip, zpip_num);
    }
}
